package com.trade.common.scheduler;

import com.trade.common.utils.Constants;
import com.trade.common.utils.IdGenerator;
import com.trade.common.utils.XaUtil;
import com.trade.module.daq.entity.*;
import com.trade.module.daq.repository.CollectRuleRepository;
import com.trade.module.daq.repository.CollectionRecordRepository;
import com.trade.module.daq.repository.TableRecordRepository;
import com.trade.module.daq.vo.TableInfoParamVo;
import com.trade.module.database.basic.ConstConfig;
import com.trade.module.database.operation.DBInfoUtil;
import com.trade.module.database.operation.DBQueryUtils;
import com.trade.module.database.meta.TableCloumnParam;
import com.trade.module.frontEndServer.FrontEndDBVo;
import lombok.extern.log4j.Log4j;
import org.apache.commons.collections.map.HashedMap;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;

import java.sql.SQLException;
import java.util.*;

/**
 * Class Description: 任务主线程
 *
 * @Version v1.0 2017/6/22 0:21
 * @JDK version 1.8.0_51
 * @Author zfb
 */
@Log4j
public class TaskRunnable implements Runnable {
    private String taskId;   //任务ID

    private String trigger;  //触发器

    private String taskType; //任务类型

    private String resId;    //资源id

    private String isIncrement; // 同步方式 y:增量更新 n:全量更新

    private long quantity = 0;

    private Long endTime;

    private SystemInfo target; //目标数据源

    private SystemInfo source; //数据源

    private List<TableInfoParamVo> inTableInfoParam; // 源表参数

    private List<TableInfoParamVo> outTableInfoParam;// 目标表参数

    private String[] inTableNameArray; // 源表名称

    private List<TableRecord> tableRecordList = new ArrayList<>();

    @Autowired
    private CollectRule collectionsRule;//规则实体

    @Autowired
    private CollectionRecordRepository collectionRecordRepository;

    @Autowired
    private CollectRuleRepository collectRuleRepository;

    @Autowired
    private TableRecordRepository tableRecordRepository;

    public TaskRunnable() {
    }

    public TaskRunnable(String resId, String trigger) {
        taskId = resId;
        this.resId = resId;
        this.trigger = trigger;
    }

    public TaskRunnable(String resId,
                        String trigger,
                        CollectRuleRepository collectRuleRepository) {
        taskId = resId;
        this.resId = resId;
        this.trigger = trigger;
        this.collectRuleRepository = collectRuleRepository;
    }

    public TaskRunnable(String[] inTableArray,
                        SystemInfo inSource,
                        SystemInfo outSource,
                        CollectRule collectionsRule,
                        List<TableInfoParamVo> inTableInfoParam,
                        List<TableInfoParamVo> outTableInfoParam,
                        CollectRuleRepository collectRuleRepository,
                        TableRecordRepository tableRecordRepository,
                        CollectionRecordRepository collectionRecordRepository) {
        taskId = collectionsRule.getId();
        this.source = inSource;
        this.target = outSource;
        this.resId = collectionsRule.getId();
        this.inTableNameArray = inTableArray;
        this.collectionsRule = collectionsRule;
        this.inTableInfoParam = inTableInfoParam;
        this.outTableInfoParam = outTableInfoParam;
        this.trigger = collectionsRule.getTimeReg();
        this.collectRuleRepository = collectRuleRepository;
        this.tableRecordRepository = tableRecordRepository;
        this.isIncrement = collectionsRule.getIsIncrement();
        this.collectionRecordRepository = collectionRecordRepository;
    }

    @Override
    public void run() {
        //调用任务执行接口
        //每次任务执行时，存储任务到collectionRecord表
        try {
            CollectRecord collectRecord = saveTaskRecord(Constants.COLLECT_ISFINSH02);
            collectionsRule.setRecordId(collectRecord.getId());
            //该规则是即时任务
            if (Constants.YesOrNo.YES.equals(collectionsRule.getIsTiming())) {
                collectionsRule.setIsFinish(Constants.COLLECT_ISFINSH02);
            }
            collectRuleRepository.save(collectionsRule);
            //配置前置机进行数据采集
            try {
                //将数据源的数据库信息写入前置机
                FrontEndDBVo frontEndDBVo = new FrontEndDBVo();
                frontEndDBVo.setDbUrl(source.getIp());
                frontEndDBVo.setDbPort(source.getPort());
                frontEndDBVo.setDbUserName(source.getUsername());
                frontEndDBVo.setDbPassword(source.getPassword());
                frontEndDBVo.setDbType(ConstConfig.DBType.valueOf(Integer.parseInt(source.getDbType())));
                frontEndDBVo.setDataTables(StringUtils.join(inTableNameArray, ",")); //使用","链接字符串
                //是否增量采集
                if (StringUtils.isNotBlank(isIncrement) && isIncrement.equals(Constants.YesOrNo.YES)) {
                    //增量采集
                    //主键集合
                    List<String> ids = new ArrayList<>();
                    Map<String, TableCloumnParam> tableCloumnParamMap = new HashMap<>();
                    for (TableInfoParamVo inTableInfoParam : inTableInfoParam) {
                        List<FieldInfo> fieldInfoList = inTableInfoParam.getFieldInfoList();
                        String ename = inTableInfoParam.getEname();
                        String keyName = null;
                        String createTime = null;
                        String modifyTime = null;
                        //判断字段是否是主键、时间戳类型
                        for (FieldInfo fieldInfo : fieldInfoList) {
                            if (Constants.YesOrNo.YES.equals(fieldInfo.getIsKey())) {
                                ids.add(fieldInfo.getEname());
                                keyName = fieldInfo.getEname();
                            }
                            if (Constants.YesOrNo.YES.equals(fieldInfo.getIsCreateTime())) {
                                createTime = fieldInfo.getEname();
                            }
                            if (Constants.YesOrNo.YES.equals(fieldInfo.getIsModifyTime())) {
                                modifyTime = fieldInfo.getEname();
                            }
                        }
                        //如果主键和时间戳为空，则不能增量采集,只能全量采集，需将目标库中的原来数据截断操作
                        //如果主键和时间戳不为空，则将主键和时间戳字段配置到前置机
                        if (StringUtils.isBlank(keyName) || StringUtils.isBlank(createTime) || StringUtils.isBlank(modifyTime)) {
                            DBQueryUtils dbQueryUtils = getDbQueryUtils();
                            for (TableInfoParamVo outTableInfoParamVo : outTableInfoParam) {
                                if (ename.equals(outTableInfoParamVo.getRelTableName())) {
                                    if (ConstConfig.DBType.ORACLE.equals(ConstConfig.DBType.valueOf(Integer.parseInt(target.getDbType())))) {
                                        dbQueryUtils.execDDLSql("truncate table " + outTableInfoParamVo.getEname());
                                    } else {
                                        dbQueryUtils.delData(outTableInfoParamVo.getEname(), null);
                                    }
                                }
                            }
                        } else {
                            tableCloumnParamMap.put(ename, new TableCloumnParam(ids, createTime, modifyTime));
                        }
                    }
                    if (tableCloumnParamMap.size() > 0) {
                        frontEndDBVo.setTableCloumnParamMap(tableCloumnParamMap);
                    }
                } else if (!isIncrement.equals(Constants.YesOrNo.YES)) {
                    //全量采集 先删除表信息
                    DBQueryUtils dbQueryUtils = getDbQueryUtils();
                    for (TableInfoParamVo outTableInfoParamVo : outTableInfoParam) {
                        if (ConstConfig.DBType.ORACLE.equals(ConstConfig.DBType.valueOf(Integer.parseInt(target.getDbType())))) {
                            dbQueryUtils.execDDLSql("truncate table " + outTableInfoParamVo.getEname());
                        } else {
                            dbQueryUtils.delData(outTableInfoParamVo.getEname(), null);
                        }
                    }
                }
                //设置请求类型：获取表数据
                frontEndDBVo.setReqType(ConstConfig.DBReqType.REQ_TABLE_DATA);

                //任务是否立即执行 (0:不立即执行，1：立即执行)
                frontEndDBVo.setExecuteImmediately(1);

                long startTime = System.currentTimeMillis();
                //创建任务处理类
                DBWorkHandler dbWorkHandler = new DBWorkHandler();
                //启动采集任务
                dbWorkHandler.startTask(frontEndDBVo, new ClientWorkCallback() {
                    @Override
                    public void workStarted(String workId) {
                        collectRecord.setWorkId(workId);
                        collectionRecordRepository.save(collectRecord);
                    }

                    @Override
                    public void workPartialData(TaskResultVo workResultVo) {
                        Map<String, List<Map<String, String>>> records = workResultVo.getFrontEndVo().getRecords();
                        try {
                            //写入数据到目标数据库
                            outputDataToDestDB(workResultVo);
                        } catch (SQLException e) {
                            e.printStackTrace();
                            log.error("写入数据到目标数据库异常："+e);
                            endTime = System.currentTimeMillis();
                            collectRecord.setCollectQuantity(quantity);
                            collectRecord.setExecuteTime((endTime - startTime) + "");
                            collectionRecordRepository.save(collectRecord);
                        }

                        Set<Map.Entry<String, List<Map<String, String>>>> entries = records.entrySet();
                        Map.Entry<String, List<Map<String, String>>> entry = (Map.Entry<String, List<Map<String, String>>>) entries.toArray()[0];
                        quantity += entry.getValue().size();
                    }

                    @Override
                    public void workFinished(TaskResultVo workResultVo) {
                        collectionsRule.setIsFinish(Constants.COLLECT_ISFINSH03);
                        collectRuleRepository.save(collectionsRule);
                        endTime = System.currentTimeMillis();
                        collectRecord.setExecuteStatus(Constants.COLLECT_ISFINSH03);
                        collectRecord.setExecuteTime((endTime - startTime) + "");
                        collectRecord.setCollectQuantity(quantity);
                        collectionRecordRepository.save(collectRecord);
                        for (TableRecord tableRecord : tableRecordList) {
                            tableRecordRepository.save(tableRecord);
                        }
                    }

                    @Override
                    public void dataQueryException(Exception e, FrontEndDBVo frontEndDBVo) {
                        log.error(e.toString() + "前置机信息：" + frontEndDBVo);
                    }
                });
            } catch (Exception e) {
                e.printStackTrace();
            }
            log.info("规则id=" + collectionsRule.getId() + ",任务id=" + " 开始执行。");
        } catch (Exception e) {
            e.printStackTrace();
            log.error("执行任务异常!", e);
        }
    }

    /**
     * 写入数据到目标数据库
     *
     * @param workResultVo 任务执行结果对象
     * @return 数据集
     */
    private Map<String, List<Map<String, String>>> outputDataToDestDB(TaskResultVo workResultVo) throws SQLException {
        /*链接目标库**/
        DBQueryUtils dbQueryUtils = getDbQueryUtils();
        /*获取开始时间**/
        long startTime = System.currentTimeMillis();
        /*获取所有的记录*/
        Map<String, List<Map<String, String>>> records = workResultVo.getFrontEndVo().getRecords();
        /*查找源库中对应的目标库的表名**/
        Set<String> tableNameSet = records.keySet();
        if(XaUtil.isEmpty(tableNameSet)){
            return null;
        }
        String srcTableName = tableNameSet.toArray(new String[0])[0]; //源表名称
        List<Map<String,String>> srcDataList = records.get(srcTableName); //源表中数据
        String outTableName = "";
        /*查找到对应的列集合**/
        List<FieldInfo> outFieldInfoList = null;
        for(TableInfoParamVo outTableInfo : outTableInfoParam){
            if(srcTableName.trim().equals(outTableInfo.getRelTableName())){
                outTableName = outTableInfo.getEname();
                outFieldInfoList = outTableInfo.getFieldInfoList();
            }
        }
        /*转化字段名称**/
        if (srcDataList == null || srcDataList.size() <= 0) {
            return null;
        }
        Map<String, String> firstDataMap = srcDataList.get(0);
        Set<String> srcCloumnNames = firstDataMap.keySet();

        Map<String, String> diffCloumnMapping = new HashedMap();
        List<String> outTableCloumnNames = new ArrayList<>();
        Map<String, String> dateMap = new HashedMap();

        if (outFieldInfoList != null) {
            for(FieldInfo fieldInfo:outFieldInfoList){
                String destCloumnName = fieldInfo.getEname();
                outTableCloumnNames.add(destCloumnName.toUpperCase());
                if (!srcCloumnNames.contains(destCloumnName)) {
                    String srcCloumnName = fieldInfo.getSourceEname();
                    diffCloumnMapping.put(srcCloumnName.toUpperCase(), destCloumnName.toUpperCase());
                }
            }
        }

        Set<String> diffCloumnKeys = diffCloumnMapping.keySet();
        List<Map<String, String>> newDataList = new ArrayList<>();
        Iterator<Map<String, String>> mapIterator = srcDataList.iterator();
        while (mapIterator.hasNext()) {
            Map<String, String> newRecordMap = new HashedMap();
            Map<String, String> recordMap = mapIterator.next();
            Set<String> srcMapKeys = recordMap.keySet();
            for (String srcCloumnName : srcMapKeys) {
                String srcCloumnNameUpper = srcCloumnName.toUpperCase();
                if (!outTableCloumnNames.contains(srcCloumnNameUpper)) {
                    if (diffCloumnKeys.contains(srcCloumnNameUpper)) {
                        String destCloumnName = diffCloumnMapping.get(srcCloumnNameUpper);
                        Object dataValue = recordMap.get(srcCloumnName);
                        String dataValueStr = null;
                        if (dataValue != null) {
                            dataValueStr = dataValue.toString();
                        }
                        newRecordMap.put(destCloumnName, dataValueStr);
                    }
                } else {
                    Object dataValue = recordMap.get(srcCloumnName);
                    String dataValueStr = null;
                    if (dataValue != null) {
                        dataValueStr = dataValue.toString();
                    }
                    newRecordMap.put(srcCloumnName, dataValueStr);
                }
            }
            newDataList.add(newRecordMap);
        }
        /*数据入库**/
        dbQueryUtils.insertBatchData(outTableName,target.getDbName(),newDataList);
        /*获取结束时间**/
        long endTime = System.currentTimeMillis();
        Set<Map.Entry<String, List<Map<String, String>>>> entries = records.entrySet();
        Map.Entry<String, List<Map<String, String>>> entry = (Map.Entry<String, List<Map<String, String>>>) entries.toArray()[0];
        long size = entry.getValue().size();

        boolean tableRecordFlag = false;
        for (TableRecord tableRecord : tableRecordList) {
            if (tableRecord.getTableName().equals(outTableName)) {
                if (!StringUtils.isNotEmpty(tableRecord.getExecuteTime())) {
                    tableRecord.setExecuteTime((endTime - startTime) + "");
                } else {
                    tableRecord.setExecuteTime(Long.parseLong(tableRecord.getExecuteTime()) + (endTime - startTime) + "");
                }
                tableRecord.setCollectQuantity(tableRecord.getCollectQuantity() + size);
                tableRecordFlag = true;
                break;
            }
        }

        if (tableRecordList.size() <= 0 || !tableRecordFlag) {
            TableRecord tableRecord = new TableRecord();
            tableRecord.setTableName(outTableName);
            tableRecord.setCollectQuantity(size);
            tableRecord.setExecuteTime((endTime - startTime) + "");
            tableRecordList.add(tableRecord);
        }

        /*记录日志**/
        log.info("数据写入目标库用时:" + "-----" + (endTime - startTime) + "ms");
        return null;
    }

    /**
     * 保存采集记录
     *
     * @param status 状态
     * @return CollectRecord 采集记录
     */
    private CollectRecord saveTaskRecord(String status) {
        CollectRecord collectionRecord = new CollectRecord();
        collectionRecord.setId(IdGenerator.createNewId());
        collectionRecord.setTaskDsc(collectionsRule.getDescription());
        collectionRecord.setExecuteTime(endTime + ""); //耗时
        collectionRecord.setTargetDBId(target.getId());
        collectionRecord.setTaskType(collectionsRule.getIsIncrement());
        collectionRecord.setExecuteStatus(status);
        collectionRecord.setTaskName(collectionsRule.getRuleName());
        collectionRecord.setTargetDBName(target.getDbName());
        return collectionRecordRepository.save(collectionRecord);
    }

    /**
     * 获取数据库操作工具
     *
     * @return DBQueryUtils
     */
    private DBQueryUtils getDbQueryUtils() {
        String driverName = DBInfoUtil.getDBDriverName(ConstConfig.DBType.valueOf(Integer.parseInt(target.getDbType())));
        String dbUrl = DBInfoUtil.getDBUrl(ConstConfig.DBType.valueOf(Integer.parseInt(target.getDbType())), target.getIp(), target.getPort(), target.getDbName());
        String username = target.getUsername();
        String password = target.getPassword();
        int connPoolSize = 20;
        return new DBQueryUtils(driverName, dbUrl, username, password, connPoolSize);
    }


    public String getTaskId() {
        return taskId;
    }

    public void setTaskId(String taskId) {
        this.taskId = taskId;
    }

    public String getTrigger() {
        return trigger;
    }

    public void setTrigger(String trigger) {
        this.trigger = trigger;
    }

    public String getTaskType() {
        return taskType;
    }

    public void setTaskType(String taskType) {
        this.taskType = taskType;
    }

    public String getResId() {
        return resId;
    }

    public void setResId(String resId) {
        this.resId = resId;
    }


    public static void main(String[] args) {
        String[] array = new String[]{"aa", "bb", "cc", "dd", "ee"};
        String join = StringUtils.join(array, ",");
        System.out.println(join);

        System.out.println("-----------------------------");

        Map<String, String> map1 = new HashMap<>();
        map1.put("a", "1");
        map1.put("b", "2");
        map1.put("c", "3");
        map1.put("d", "4");
        map1.put("e", "500");
        Map.Entry<String, String> entity = (Map.Entry<String, String>) map1.entrySet().toArray()[4];
        System.out.println(entity.getValue());

        System.out.println("-----------------------------");

        Set<String> set = new HashSet<>();
        set.add("1");
        set.add("2");
        set.add("3");
        set.add("4");
        //如果new String[] 数组能容纳set则返回new String[],否则返回一个包含set的新数组
        String[] strings = set.toArray(new String[0]); //返回一个包含set的新数组
        System.out.println(strings[3]);
    }
}
